package com.zidongxiangxi.rabbit.starter.job;

import com.zidongxiangxi.rabbit.core.exception.ReliableRabbitException;
import com.zidongxiangxi.rabbit.core.job.AbstractDaemonWorkThread;
import com.zidongxiangxi.rabbit.core.job.AbstractWorkThread;
import com.zidongxiangxi.rabbit.starter.util.ZookeeperUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 工作线程的守护进程
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class ZookeeperDaemonWorkThread extends AbstractDaemonWorkThread {
    private final Logger logger = LoggerFactory.getLogger(ZookeeperDaemonWorkThread.class);

    private static final int MAX_SHUTDOWN_RETRY_TIMES = 3;
    private static final String WORK_THREAD_LOCK_KEY = "/consume/work_thread_lock_key";
    private final ExecutorService WORK_THREAD_POOL = new ThreadPoolExecutor(64, 64, 10, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1), new CustomizedThreadFactory());
    private final AtomicBoolean stop = new AtomicBoolean(false);
    private final String applicationName;
    private final CuratorFramework client;
    private final List<AbstractWorkThread> waitingThreads = new LinkedList<>();
    private final List<AbstractWorkThread> workingThreads = new LinkedList<>();

    private volatile boolean hasGetLock = false;

    public ZookeeperDaemonWorkThread(
            String applicationName,
            String zookeeperAddress
    ) {
        this.applicationName = applicationName;
        this.client = ZookeeperUtils.createCuratorFramework(zookeeperAddress, 20, 20);
    }

    @Override
    public void run() {
        InterProcessMutex lock = getLock();
        if (Objects.isNull(lock)) {
            throw new RuntimeException("fail to get distribution lock");
        }
        //获取分布式锁
        try {
            boolean isGet = tryLock(lock);
            if (!isGet) {
                return;
            }
        } catch (Exception e) {
            return;
        }

        logger.info("work thread success to get distribution lock......");
        hasGetLock = true;

        // 启动等待中的任务
        synchronized (this) {
            if (!CollectionUtils.isEmpty(waitingThreads)) {
                for (AbstractWorkThread workThread : waitingThreads) {
                    WORK_THREAD_POOL.submit(workThread);
                }
                // 全部转移到工作中的数组
                workingThreads.addAll(waitingThreads);
                waitingThreads.clear();
            }
        }


        try {
            while (true) {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    logger.warn("work thread is be interrupted", e);
                }
                if (stop.get()) {
                    break;
                }
                // TODO 检查真正工作的两个线程是否活着，如果线程挂了，可以重新启动或者跳出循环
            }
        } finally {
            logger.info("work thread should be stopped, and trying to release distribution lock......");
            unLock(lock);
            ZookeeperUtils.closeCuratorFramework(client);
            hasGetLock = false;
        }

    }

    @Override
    public synchronized void addWorkThread(Collection<AbstractWorkThread> workThreads) {
        if (stop.get()) {
            throw new ReliableRabbitException("daemon work thread is stopped......");
        }
        if (CollectionUtils.isEmpty(workThreads)) {
            return;
        }
        if (hasGetLock) {
            workingThreads.addAll(workThreads);
            for (AbstractWorkThread workThread : workThreads) {
                WORK_THREAD_POOL.submit(workThread);
            }
        } else {
            waitingThreads.addAll(workThreads);
        }
    }

    /**
     * 停止工作线程
     */
    @Override
    public synchronized void stopWorking() {
        if (!stop.compareAndSet(false, true)) {
            logger.warn("daemon work thread is topped, need not stop again......");
            return;
        }
        for (AbstractWorkThread workThread : workingThreads) {
            workThread.stop();
        }
        WORK_THREAD_POOL.shutdownNow();
        int retryTimes = 0;
        while (retryTimes < MAX_SHUTDOWN_RETRY_TIMES && !WORK_THREAD_POOL.isTerminated()) {
            retryTimes++;
            try {
                WORK_THREAD_POOL.awaitTermination(3, TimeUnit.SECONDS);
            } catch (InterruptedException ignore) {}
        }
        this.interrupt();
    }

    private InterProcessMutex getLock() {
        if (Objects.nonNull(client)) {
            return ZookeeperUtils.getLock(client,
                    applicationName + WORK_THREAD_LOCK_KEY);
        }
        return null;
    }

    private Boolean tryLock(InterProcessMutex lock) throws Exception{
        try {
            while (!lock.acquire(60, TimeUnit.SECONDS)) {
            }
        } catch (Exception e) {
            if (e instanceof InterruptedException) {
                throw e;
            }
            logger.error("fail to get distribution lock", e);
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }

    private void unLock(InterProcessMutex lock) {
        try {
            lock.release();
        } catch (Throwable e){
            logger.error("fail to release distribution lock", e);
        }
    }

    /**
     * 自定义线程工厂,新的线程利用该类进行创建
     */
    static class CustomizedThreadFactory implements ThreadFactory {
        private final AtomicInteger sequence = new AtomicInteger(1);
        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable);
            thread.setName("work-thread-" + sequence.getAndIncrement());
            return thread;
        }
    }
}
